package com.itheima.search.listener;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.DeleteRequest;
import co.elastic.clients.elasticsearch.core.DeleteResponse;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.itheima.search.domain.NewsDocument;

import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;

/**
 * 监听新闻微服务的消息，同步到Elasticsearch
 * @author NicelYHHello
 */
@Component
@Slf4j
public class MqToEsListener {

    @Autowired
    private ElasticsearchClient elasticsearchClient;
    // 新增、修改、删除队列
    public static final String NEWS_ADD_QUEUE = "news.add.queue";
    public static final String NEWS_UPDATE_QUEUE = "news.update.queue";
    public static final String NEWS_DELETE_QUEUE = "news.delete.queue";

    // 日期格式化器
    private static final DateTimeFormatter DATE_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    /**
     * 监听新增队列，接收新闻文档消息并新增到Elasticsearch
     * @param newsDocument 接收到的新闻文档
     */
    @RabbitListener(queues = NEWS_ADD_QUEUE)
    public void listenInsert(NewsDocument newsDocument) {
        log.info("消费者接收到新增消息：{}", newsDocument);

        try {
            // 将NewsDocument转换为Map格式，确保字段类型与ES索引匹配
            Map<String, Object> documentMap = convertToMap(newsDocument);

            // 创建索引请求
            IndexRequest<Map<String, Object>> indexRequest = IndexRequest.of(b -> b
                    .index("tb_news")
                    .id(newsDocument.getNewsId().toString())
                    .document(documentMap)
            );

            // 执行索引操作
            IndexResponse response = elasticsearchClient.index(indexRequest);
            log.info("新闻新增到ES成功，文档ID: {}", response.id());

        } catch (Exception e) {
            log.error("新闻新增到ES失败: {}", e.getMessage(), e);
            // 这里可以根据业务需求添加重试机制或记录到失败表
        }
    }

    /**
     * 监听修改队列，接收新闻文档消息并更新到Elasticsearch
     * @param newsDocument 接收到的新闻文档
     */
    @RabbitListener(queues = NEWS_UPDATE_QUEUE)
    public void listenUpdate(NewsDocument newsDocument) {
        log.info("消费者接收到修改消息：{}", newsDocument);

        try {
            // 将NewsDocument转换为Map格式，确保字段类型与ES索引匹配
            Map<String, Object> documentMap = convertToMap(newsDocument);

            // 创建索引请求
            IndexRequest<Map<String, Object>> indexRequest = IndexRequest.of(b -> b
                    .index("tb_news")
                    .id(newsDocument.getNewsId().toString())
                    .document(documentMap)
            );

            // 执行索引操作
            IndexResponse response = elasticsearchClient.index(indexRequest);
            log.info("新闻更新到ES成功，文档ID: {}", response.id());

        } catch (Exception e) {
            log.error("新闻更新到ES失败: {}", e.getMessage(), e);
            // 这里可以根据业务需求添加重试机制或记录到失败表
        }
    }

    /**
     * 监听删除队列，接收新闻ID并删除ES中的文档
     * @param newsIds 接收到的新闻ID
     */
    @RabbitListener(queues = NEWS_DELETE_QUEUE)
    public void listenDelete(Long[] newsIds) {
        log.info("消费者接收到删除消息，新闻ID: {}", newsIds);
        for (Long newsId : newsIds) {
            try {
                // 创建删除请求
                DeleteRequest deleteRequest = DeleteRequest.of(b -> b
                        .index("tb_news")
                        .id(newsIds.toString())
                );

                // 执行删除操作
                DeleteResponse response = elasticsearchClient.delete(deleteRequest);

                if ("NotFound".equals(response.result().name())) {
                    log.warn("要删除的文档不存在，新闻ID: {}", newsIds);
                } else {
                    log.info("新闻从ES删除成功，文档ID: {}", response.id());
                }

            } catch (Exception e) {
                log.error("新闻从ES删除失败: {}", e.getMessage(), e);
                // 这里可以根据业务需求添加重试机制或记录到失败表
            }
        }

    }

    /**
     * 将NewsDocument转换为Map格式，确保字段类型与ES索引匹配
     */
    private Map<String, Object> convertToMap(NewsDocument newsDocument) {
        Map<String, Object> map = new HashMap<>();

        map.put("newsId", newsDocument.getNewsId());
        map.put("title", newsDocument.getTitle());
        map.put("typeId", newsDocument.getTypeId());
        map.put("content", newsDocument.getContent());
        map.put("delayTime", newsDocument.getDelayTime());
        map.put("source", newsDocument.getSource());
        map.put("createBy", newsDocument.getCreateBy());

        // 格式化日期字段
        if (newsDocument.getCreateTime() != null) {
            map.put("createTime", newsDocument.getCreateTime().format(DATE_FORMATTER));
        }

        map.put("updateBy", newsDocument.getUpdateBy());

        if (newsDocument.getUpdateTime() != null) {
            map.put("updateTime", newsDocument.getUpdateTime().format(DATE_FORMATTER));
        }

        map.put("status", newsDocument.getStatus());
        map.put("remark", newsDocument.getRemark());
        map.put("noticeId", newsDocument.getNoticeId());

        return map;
    }

    /**
     * 检查索引是否存在，如果不存在则创建
     */
    public void ensureIndexExists() {
        try {
            // 检查索引是否存在
            boolean exists = elasticsearchClient.indices().exists(e -> e.index("tb_news")).value();

            if (!exists) {
                log.warn("索引 tb_news 不存在，请先手动创建索引");
                log.info("可以使用以下命令创建索引:");
                log.info("PUT /tb_news");
                log.info("{}", getIndexDefinition());
            } else {
                log.info("索引 tb_news 已存在");
            }
        } catch (Exception e) {
            log.error("检查索引失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 获取索引定义（用于日志输出）
     */
    private String getIndexDefinition() {
        return "{\n" +
                "  \"settings\": {\n" +
                "    \"number_of_shards\": 3,\n" +
                "    \"number_of_replicas\": 1,\n" +
                "    \"analysis\": {\n" +
                "      \"analyzer\": {\n" +
                "        \"ik_smart_analyzer\": {\n" +
                "          \"type\": \"custom\",\n" +
                "          \"tokenizer\": \"ik_smart\"\n" +
                "        },\n" +
                "        \"ik_max_word_analyzer\": {\n" +
                "          \"type\": \"custom\",\n" +
                "          \"tokenizer\": \"ik_max_word\"\n" +
                "        }\n" +
                "      }\n" +
                "    }\n" +
                "  },\n" +
                "  \"mappings\": {\n" +
                "    \"properties\": {\n" +
                "      \"newsId\": {\n" +
                "        \"type\": \"long\"\n" +
                "      },\n" +
                "      \"title\": {\n" +
                "        \"type\": \"text\",\n" +
                "        \"analyzer\": \"ik_max_word_analyzer\",\n" +
                "        \"search_analyzer\": \"ik_smart_analyzer\",\n" +
                "        \"fields\": {\n" +
                "          \"keyword\": {\n" +
                "            \"type\": \"keyword\"\n" +
                "          }\n" +
                "        }\n" +
                "      },\n" +
                "      \"typeId\": {\n" +
                "        \"type\": \"integer\"\n" +
                "      },\n" +
                "      \"content\": {\n" +
                "        \"type\": \"text\",\n" +
                "        \"analyzer\": \"ik_max_word_analyzer\",\n" +
                "        \"search_analyzer\": \"ik_smart_analyzer\"\n" +
                "      },\n" +
                "      \"delayTime\": {\n" +
                "        \"type\": \"byte\"\n" +
                "      },\n" +
                "      \"source\": {\n" +
                "        \"type\": \"keyword\"\n" +
                "      },\n" +
                "      \"createBy\": {\n" +
                "        \"type\": \"keyword\"\n" +
                "      },\n" +
                "      \"createTime\": {\n" +
                "        \"type\": \"date\",\n" +
                "        \"format\": \"yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis\"\n" +
                "      },\n" +
                "      \"updateBy\": {\n" +
                "        \"type\": \"keyword\"\n" +
                "      },\n" +
                "      \"updateTime\": {\n" +
                "        \"type\": \"date\",\n" +
                "        \"format\": \"yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis\"\n" +
                "      },\n" +
                "      \"status\": {\n" +
                "        \"type\": \"short\"\n" +
                "      },\n" +
                "      \"remark\": {\n" +
                "        \"type\": \"text\",\n" +
                // 修正了这里的语法错误，添加了缺失的逗号
                "        \"analyzer\": \"ik_max_word_analyzer\"\n" +
                "      },\n" +
                "      \"noticeId\": {\n" +
                "        \"type\": \"integer\"\n" +
                "      }\n" +
                "    }\n" +
                "  }\n" +
                "}";
    }
}